/************************************
 * Package: RGcpp					*
 * Authors: Rui Eduardo Gouveia Gil	*
 * Date: 2010						*
 ************************************/
#include "rgcpp/net/sockets/TCPSocketPool.h"
#include "rgcpp/net/sockets/TCPSocket.h"

namespace rgcpp
{

TCPSocketPool::TCPSocketPool()
{
	timeout.tv_usec = NO_TIMEOUT;
	timeout.tv_sec = NO_TIMEOUT;

#ifdef _WIN32
	if(::_pipe(pipeFD,256, O_BINARY) == -1)
#else
	if(::pipe(pipeFD) == -1)
#endif
	{
		Throw(SocketException(std::string("Error opening the pipe")+strerror(errno)));
	}

	running = true;
}

TCPSocketPool::~TCPSocketPool()
{
#ifdef _WIN32
	_close(pipeFD[0]);
	_close(pipeFD[1]);
#else
	close(pipeFD[0]);
	close(pipeFD[1]);
#endif
}

void TCPSocketPool::addSocket(TCPSocket* socket)
{
	boost::mutex::scoped_lock lock(mut);
	socketFDList[socket->getSocketFD()] = Types<TCPSocket*,bool>(socket,true);

#ifdef _WIN32
	::_write(pipeFD[1],"\0",1);
#else
	if(::write(pipeFD[1],"\0",1)==-1)
	{
		Throw(SocketException("Error writing to pipe"));
	}
#endif
}

TCPSocket* TCPSocketPool::getActiveSocket()
{
	boost::mutex::scoped_lock lock(mutRead);

	if(!running)
		return NULL;

	int value;

	while(true)
	{
		FD_ZERO(&fdList);
		FD_SET(pipeFD[0], &fdList);
		maxFD = pipeFD[0];
		{
			boost::mutex::scoped_lock lock(mut);
			for(Map<int,Types<TCPSocket*,bool> >::iterator i=socketFDList.begin(); i!=socketFDList.end(); i++)
			{
				if(i->second.t2)
				{
					FD_SET(i->first, &fdList);
					if(i->first>maxFD)
						maxFD = i->first;
				}
			}
		}
		maxFD++;
		struct timeval timeout = this->timeout;
		do
		{
			if(!timeout.tv_usec && !timeout.tv_sec)
				value = select(maxFD,&fdList,NULL,NULL,NULL);
			else
				value = select(maxFD,&fdList,NULL,NULL,&timeout);
		}while(value == -1&& errno == EINTR);

		if(value == 0)
		{
			Throw(SocketException("Error reading from socket pool - Timeout exception"));
		}
		else if(value == -1)
		{
			Throw(SocketException(std::string("Error reading from socket pool")+strerror(errno)));
		}
		else if(FD_ISSET(pipeFD[0],&fdList))
		{
#ifdef _WIN32
			::_read(pipeFD[0],pipeBuffer,1);
#else
			if(::read(pipeFD[0],pipeBuffer,1)==-1)
			{
				Throw(SocketException("Error reading from pipe"));
			}
#endif

			if(pipeBuffer[0])
				return NULL;
		}
		else
			break;
	}

	TCPSocket* socket=NULL;
	for(Map<int,Types<TCPSocket*,bool> >::iterator i=socketFDList.begin(); i!=socketFDList.end(); i++)
	{
		if(FD_ISSET(i->first,&fdList))
		{
			socket = i->second.t1;
			i->second.t2 = false;
			break;
		}
	}

	return socket;
}

void TCPSocketPool::releaseSocket(TCPSocket* socket)
{
	boost::mutex::scoped_lock lock(mut);
	socketFDList[socket->getSocketFD()].t2 = true;

#ifdef _WIN32
	::_write(pipeFD[1],"\0",1);
#else
	if(::write(pipeFD[1],"\0",1)==-1)
	{
		Throw(SocketException("Error writing to pipe"));
	}
#endif
}

void TCPSocketPool::removeSocket(TCPSocket* socket)
{
	boost::mutex::scoped_lock lock(mut);

	if(socketFDList.find(socket->getSocketFD())==socketFDList.end())
		return;

	socketFDList.erase(socket->getSocketFD());

#ifdef _WIN32
	::_write(pipeFD[1],"\0",1);
#else
	if(::write(pipeFD[1],"\0",1)==-1)
	{
		Throw(SocketException("Error writing to pipe"));
	}
#endif
}

void TCPSocketPool::setTimeout(const unsigned long timeout)
{
	this->timeout.tv_usec = (timeout % 1000) * 1000;
	this->timeout.tv_sec = timeout / 1000;
}

unsigned long TCPSocketPool::getTimeout()
{
	unsigned long timeout;

	timeout = this->timeout.tv_sec * 1000;
	timeout += this->timeout.tv_usec / 1000;

	return timeout;
}

void TCPSocketPool::release()
{
	running = false;

#ifdef _WIN32
	::_write(pipeFD[1],"\1",1);
#else
	if(::write(pipeFD[1],"\1",1)==-1)
	{
		Throw(SocketException("Error writing to pipe"));
	}
#endif
}

}
